package com.ibm.cps.kafka

import java.util.Properties

import com.ibm.util.LocalConfig
import kafka.admin.AdminUtils
import kafka.utils.ZKStringSerializer
import org.I0Itec.zkclient.ZkClient
import org.apache.log4j.Logger

/**
 * Created by telekinesis on 10/13/15.
 */
object KafkaTopicUtils {
  val logger: Logger = Logger.getLogger(KafkaTopicUtils.getClass)

  def createTopic(zookeeperHost: String, topicName: String, numPartitions: Int, replicationFactor: Int): Unit = {
    val sessionTimeoutMs = 10000
    val connectionTimeoutMs = 10000
    val zkClient = new ZkClient(zookeeperHost, sessionTimeoutMs, connectionTimeoutMs,
      ZKStringSerializer)

    val topicConfig = new Properties
    if (!AdminUtils.topicExists(zkClient, topicName)) {
      logger.info("creating kafka topic: " + topicName + " with " + numPartitions + " partitions, " + replicationFactor + " replicas")
      AdminUtils.createTopic(zkClient, topicName, numPartitions, replicationFactor, topicConfig)
    } else {

    }
    zkClient.close();
  }

  def removeTopic(zookeeperHost: String, topicName: String): Unit = {
    val sessionTimeoutMs = 10000
    val connectionTimeoutMs = 10000
    val zkClient = new ZkClient(zookeeperHost, sessionTimeoutMs, connectionTimeoutMs,
      ZKStringSerializer)
    logger.info("deleting kafka topic: " + topicName)
    AdminUtils.deleteTopic(zkClient, topicName)
    zkClient.close();
  }

  def main(args: Array[String]) {
    val zookeeperHost = LocalConfig.ZOOKEEPER_HOST_PORT
    val topicName = "TEST_TOPIC_123"
    val partitions = 1
    val replications = 1
    createTopic(zookeeperHost, topicName, partitions, replications)
    createTopic(zookeeperHost, topicName, partitions, replications)
    removeTopic(zookeeperHost, topicName)
  }
}
